package com.tmsb.utils.trigger;

import com.tmsb.utils.ConfUtil;
import lombok.var;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 *Description:自定义Trigger,默认5分钟输出一次
 * Trigger<T, TimeWindow> 输入类型，窗口类型
 *Version:1.0.0
 *@author shihb
 *@date 2020/6/12 11:39
 */
public class TimerTrigger<IN> extends Trigger<IN, TimeWindow> {

  private Long triggerTime=0L;
//  private int intervalSecond;
//
//  public TimerTrigger() {
//    this(5*60);
//  }
//
//  public TimerTrigger(int intervalSecond) {
//    this.intervalSecond = intervalSecond;
//  }

  @Override
  public  TriggerResult onElement(IN in, long l, TimeWindow window,
      TriggerContext ctx) throws Exception {
    var currentWatermark = ctx.getCurrentWatermark();
    if (window.maxTimestamp() <= currentWatermark) {
      return TriggerResult.FIRE_AND_PURGE;
    }
    if((currentWatermark-triggerTime)/1000>= ConfUtil.getTriggerInterval()){
      triggerTime=currentWatermark;
      return TriggerResult.FIRE_AND_PURGE;
    }else{
      return TriggerResult.CONTINUE;
    }

  }

  @Override
  public TriggerResult onProcessingTime(long l, TimeWindow timeWindow,
      TriggerContext triggerContext) throws Exception {
    return TriggerResult.CONTINUE;
  }

  @Override
  public TriggerResult onEventTime(long l, TimeWindow timeWindow, TriggerContext triggerContext)
      throws Exception {
    return TriggerResult.CONTINUE;
  }

  @Override
  public void clear(TimeWindow timeWindow, TriggerContext triggerContext) throws Exception {

  }
}
